package com.yyfax.pay.quartz.jobs;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.commons.collections.CollectionUtils;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.yyfax.framework.common.Log;
import com.yyfax.framework.spring.ApplicationContextHolder;
import com.yyfax.pay.common.constants.QuartzConstant;
import com.yyfax.pay.common.enums.TaskStateEnum;
import com.yyfax.pay.depository.dao.FuiouNotifyTaskDao;
import com.yyfax.pay.depository.domain.FuiouNotifyTask;
import com.yyfax.pay.quartz.service.FuiouDepositNotifyService;

/**
 * 富友存管异步通知任务（扫描机制）
 * 
 * @author buyi
 * @date 2017-09-04 11:06:03
 * @since 2.0.8
 */
@Component
public class FuiouDepositNotifyFixJob implements Job {

	private static final Logger logger = LoggerFactory.getLogger(FuiouDepositNotifyFixJob.class);

	private static FuiouNotifyTaskDao fuiouNotifyTaskDao;

	private static FuiouDepositNotifyService fuiouDepositNotifyService;

	@Resource
	private QuartzConstant quartzConstant;

	private ExecutorService executors;

	@PostConstruct
	private void init() {
		fuiouNotifyTaskDao = ApplicationContextHolder.getBean(FuiouNotifyTaskDao.class);
		fuiouDepositNotifyService = ApplicationContextHolder.getBean(FuiouDepositNotifyService.class);

		// 主要为IO请求，线程池大小设置为2*CPU个数+1
		int coreSize = Runtime.getRuntime().availableProcessors() * 2 + 1;
		this.executors = new ThreadPoolExecutor(coreSize, quartzConstant.getQuartzPoolMaxSize(), quartzConstant.getQuartzPoolKeepAlive(),
				TimeUnit.HOURS, new LinkedBlockingQueue<Runnable>(quartzConstant.getQuartzPoolQueueSize()),
				new FuioudepositNotifyRejectedExecutionHandler());
	}

	@Override
	public void execute(JobExecutionContext context) throws JobExecutionException {
		String op = "FuiouDepositNotifyJob.execute";
		logger.info(Log.op(op).msg("执行富友存管异步通知任务").kv("JobExecutionContext", context).toString());

		List<FuiouNotifyTask> tasks = fuiouNotifyTaskDao.selectActiveList(quartzConstant.getQueryPageSize());
		if (CollectionUtils.isEmpty(tasks)) {
			logger.info(Log.op(op).msg("【富友存管】异步通知没有可执行记录").toString());
			return;
		}

		List<Integer> ids = Lists.transform(tasks, new Function<FuiouNotifyTask, Integer>() {

			@Override
			public Integer apply(FuiouNotifyTask input) {
				return input.getId();
			}

		});

		// 更新状态为处理中
		fuiouNotifyTaskDao.updateStateList(TaskStateEnum.DEALING.getValue(), ids);

		handleList(tasks);
	}

	/**
	 * 执行集合
	 *
	 * @author buyi
	 * @date 2018-01-23 15:22:42
	 * @since v2.0.4
	 * @param tasks
	 */
	private void handleList(List<FuiouNotifyTask> tasks) {
		for (final FuiouNotifyTask task : tasks) {
			executors.execute(FuioudepositNotifyThread.newInstance(task));
		}
	}

	/**
	 * 富友异步通知线程类，用于线程池处理
	 * 
	 * @author buyi
	 * @date 2018-01-25 10:14:14
	 * @since v2.0.8
	 */
	private static class FuioudepositNotifyThread implements Runnable {
		private FuiouNotifyTask task;

		public FuioudepositNotifyThread(FuiouNotifyTask task) {
			super();
			this.task = task;
		}

		public static FuioudepositNotifyThread newInstance(FuiouNotifyTask task) {
			return new FuioudepositNotifyThread(task);
		}

		public FuiouNotifyTask getTask() {
			return task;
		}

		@Override
		public void run() {
			fuiouDepositNotifyService.executeTask(task);
		}
	}

	/**
	 * 富友存管异步通知任务线程池队列拒绝策略
	 * 
	 * @author buyi
	 * @date 2018-01-25 10:46:31
	 * @since v2.0.8
	 */
	private static class FuioudepositNotifyRejectedExecutionHandler implements RejectedExecutionHandler {
		private static final Logger logger = LoggerFactory.getLogger(FuioudepositNotifyRejectedExecutionHandler.class);

		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			final String op = "FuioudepositThreadPoolRejectedExecutionHandler.rejectedExecution";
			if (!(r instanceof FuioudepositNotifyThread)) {
				// 如果线程类型不是富友存管通知类型，则不执行
				logger.warn(Log.op(op).msg("【富友存管】【异步通知业务】添加异步通知任务到线程池，任务队列已满，线程类型非指定FuioudepositNotifyThread类型").toString());
				return;
			}

			FuioudepositNotifyThread fuioudepositNotifyThread = (FuioudepositNotifyThread) r;
			FuiouNotifyTask task = fuioudepositNotifyThread.getTask();

			logger.warn(Log.op(op).msg("【富友存管】【异步通知业务】添加异步通知任务到线程池，任务队列已满，重置任务状态").kv("id", task.getId()).kv("busiId", task.getBusiId()).toString());

			// 如果添加线程池失败，则更新任务状态为初始化状态，等待下一次轮询
			fuiouNotifyTaskDao.updateStateById(task.getId(), TaskStateEnum.ACTIVE.getValue());
		}
	}
	
	public static void main(String[] args) {
		FuioudepositNotifyThread thread1 = new FuioudepositNotifyThread(new FuiouNotifyTask());
		System.out.println(thread1);
		FuioudepositNotifyThread thread2 = new FuioudepositNotifyThread(new FuiouNotifyTask());
		System.out.println(thread2);
		System.out.println(thread1 == thread2);
	}
}
